/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.client;

import java.io.IOException;
import java.util.List;
import java.util.Properties;

import org.sss.common.Reflection;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.HashPartitioner;
import org.sss.mapreduce.KeyFormat;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Partitioner;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.datatype.Packable;

/**
 * This is wrapper of JobEngine to execute Simple Map-Reduce job simply.
 *
 * This class can be used to submit one input, one Mapper, 
 * one Reducer and one output job.
 */
public class SimpleJob {
  private JobEngine engine;
  private Class<? extends Mapper> mapper = null;
  private Class<? extends Reducer> combiner = null;
  private Class<? extends Reducer> reducer = null;
  private Class<? extends Partitioner> partitioner = HashPartitioner.class;
  private Class<? extends Packable> inputKeyClass = null;
  private Class<? extends Packable> inputValueClass = null;
  private Class<? extends Packable> outputKeyClass = null;
  private Class<? extends Packable> outputValueClass = null;
  private Class<? extends Packable> mapOutputKeyClass = null;
  private Class<? extends Packable> mapOutputValueClass = null;
  private GroupID input  = null;
  private GroupID output = null;
  private final String name = "Simple Job";
  private Job mapperJob;
  private Job reducerJob;
  private Properties execInfo = null;
  
  /**
   * Construct this class.
   *
   * @param client the @{link SssClient}
   */
  public SimpleJob(SssClient client) throws SssException {
    engine = new JobEngine(client);
  }

  /**
   * Executes this job.
   *
   * This method does not return while job is running.
   *
   * @return @{link GroupID} of job-output.
   */
  public GroupID execute() throws SssException {
    assert mapper != null: "mapper is null";
    assert reducer != null: "reducer is null";
    assert input != null: "input is null";

    try {
      Configuration conf = engine.getConfiguration();

      KeyFormat tmpKeyFmt = conf.getKeyFormat("map.output.keyformat",    "job.new_group.keyformat");
      KeyFormat outKeyFmt = conf.getKeyFormat("reduce.output.keyformat", "job.new_group.keyformat");
      int nSlot = conf.getInt("job.new_group.numslot", 16);

      GroupID commTmp = GroupID.createRandom(tmpKeyFmt, false, false, nSlot);
      GroupID commOut = (output != null)? output : GroupID.createRandom(outKeyFmt, false, false, nSlot);

      mapperJob = engine.getJobBuilder(name + "Mapper", mapper)
        .addInput("input", input, Job.keyClass(inputKeyClass), Job.valueClass(inputValueClass))
        .addOutput("tmpdata", commTmp, combiner, Job.partitioner(partitioner), 
                                                 Job.keyClass(mapOutputKeyClass),
                                                 Job.valueClass(mapOutputValueClass))
        .build();
      
      reducerJob = engine.getJobBuilder(name + "Reducer", reducer)
        .addInput("tmpdata", commTmp, Job.keyClass(mapOutputKeyClass),
                                      Job.valueClass(mapOutputValueClass))
        .addOutput("output", commOut, Job.partitioner(partitioner),
                                      Job.keyClass(outputKeyClass),
                                      Job.valueClass(outputValueClass))
        .build();

      execInfo = engine.exec();

      return commOut;
    } finally {
      engine.dispose();
      engine = null;
    }
  }

  /**
   * Get mapper class.
   */
  public Class<?> getMapperClass() {
    return mapper;
  }

  /**
   * Set mapper class.
   *
   * NOTE: User must set Mapper class using this method, before call @{execute} method.
   */
  public void setMapperClass(Class<? extends Mapper> mapper) {
    // TODO: Check whether mapper is 1input-1output mapper.
    this.mapper = mapper;
  }

  /**
   * Get Reducer class.
   */
  public Class<?> getReducerClass() {
    return reducer;
  }

  /**
   * Set Reducer class.
   *
   * NOTE: User must set Reducer class using this method, before call @{execute} method.
   */
  public void setReducerClass(Class<? extends Reducer> reducer) {
    // TODO: Check whether reducer is 1input-1output reducer.
    this.reducer = reducer;
  }

  /**
   * Get Combiner class.
   */
  public Class<? extends Reducer> getCombinerClass() {
    return combiner;
  }

  /**
   * Set combiner class.
   */
  public void setCombinerClass(Class<? extends Reducer> combiner) throws SssException {
    if (!Reflection.createProcInfo(combiner).isAvailableAsCombiner()) {
      throw new SssException(combiner.getName() + " is invalid combiner class.");
    }
    this.combiner = combiner;
  }

  /**
   * Get Partitioner class.
   */
  public Class<? extends Partitioner> getPartitionerClass() {
    return partitioner;
  }

  /**
   * Set Partitioner class.
   *
   * When Partitioner class is not specified, @{link HashPartitioner} is used.
   */
  public void setPartitionerClass(Class<? extends Partitioner> partitioner) {
    this.partitioner = partitioner;
  }

  /**
   * Get input TupleGroup ID.
   */
  public GroupID getInputGID() {
    return input;
  }

  /**
   * Set input TupleGroup ID.
   *
   * NOTE: User must set input TupleGroup ID using this method, before call @{execute} method.
   */
  public void setInputGID(GroupID gid) {
    this.input = gid;
  }

  /**
   * Set output TupleGroup ID.
   *
   * NOTE: User must set input TupleGroup ID using this method, before call @{execute} method.
   */
  public void setOutputGID(GroupID gid) {
    this.output = gid;
  }

  
  /**
   * Get the {@link Configuration}.
   */
  public Configuration getConfiguration() {
    return engine.getConfiguration();
  }

  /**
   * Register a jar file to this <code>JobEngine</code>
   *
   * @param   fileName  path to a jar file
   * @throws  IOException
   */
  public void registerJarFile(String fileName) throws IOException {
    engine.registerJarFile(fileName);
  }

  /**
   * Register jar files to this <code>JobEngine</code>
   *
   * @param   jarFiles  collection of <code>String</code> which represent path to a jar file
   * @throws  IOException
   */
  public void registerJarFile(List<String> jarFiles) throws IOException {
    engine.registerJarFile(jarFiles);
  }

  public Long getMapperCounoter(Enum<?> counterName) {
    return mapperJob.getCounter(counterName);
  }

  public Long getReducerCounoter(Enum<?> counterName) {
    return reducerJob.getCounter(counterName);
  }

  /**
   * Get execution information.
   *
   * NOTE: this returns null before {@link SimpleJob#execute} has been done.
   */
  public Properties getExecInfo() {
    return execInfo;
  }

  public Class<? extends Packable> getInputKeyClass() {
    return inputKeyClass;
  }
  public void setInputKeyClass(Class<? extends Packable> inputKeyClass) {
    this.inputKeyClass = inputKeyClass;
  }

  public Class<? extends Packable> getInputValueClass() {
    return inputValueClass;
  }
  public void setInputValueClass(Class<? extends Packable> inputValueClass) {
    this.inputValueClass = inputValueClass;
  }

  public Class<? extends Packable> getOutputKeyClass() {
    return outputKeyClass;
  }
  public void setOutputKeyClass(Class<? extends Packable> outputKeyClass) {
    this.outputKeyClass = outputKeyClass;
  }

  public Class<? extends Packable> getOutputValueClass() {
    return outputValueClass;
  }
  public void setOutputValueClass(Class<? extends Packable> outputValueClass) {
    this.outputValueClass = outputValueClass;
  }

  public Class<? extends Packable> getMapOutputKeyClass() {
    return mapOutputKeyClass;
  }
  public void setMapOutputKeyClass(Class<? extends Packable> mapOutputKeyClass) {
    this.mapOutputKeyClass = mapOutputKeyClass;
  }

  public Class<? extends Packable> getMapOutputValueClass() {
    return mapOutputValueClass;
  }
  public void setMapOutputValueClass(Class<? extends Packable> mapOutputValueClass) {
    this.mapOutputValueClass = mapOutputValueClass;
  }
}
